package com.ljm.reactor.Sinks;

import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;

public class BufferExample {
    public static void main(String[] args) throws InterruptedException {
        // 使用 Flux.interval 创建两个生产者，分别每 100 毫秒和每 150 毫秒生成一条消息，并生成 10 条消息后停止。
        // 创建一个Flux，产生一些消息
        Flux<String> producer1 = Flux.interval(Duration.ofMillis(100))
                .map(i -> "Producer 1 - Message " + i)
                .take(10); // 生成10条消息

        Flux<String> producer2 = Flux.interval(Duration.ofMillis(150))
                .map(i -> "Producer 2 - Message " + i)
                .take(10); // 生成10条消息

        // 合并两个生产者的消息流
        Flux<String> mergedFlux = Flux.merge(producer1, producer2);

        // 使用buffer操作符按每5条消息进行分批
        Flux<List<String>> bufferedFlux = mergedFlux.buffer(Duration.ofSeconds(1));

        // 订阅并消费批量消息
        bufferedFlux.subscribe(batch -> {
            System.out.println("Received a batch of messages:");
            batch.forEach(System.out::println);
        });

        // 等待足够的时间以便查看输出结果
        Thread.sleep(5000);
    }
}

